package syR

import (
	"db2s/Fp"
	"db2s/Meta"
	"db2s/arg"
	ea "db2s/encryptionAlgorithm"
	"db2s/full"
	"db2s/global"
	"db2s/go-log/log"
	"db2s/outPut"
	"db2s/parDef"
	"db2s/ref"
	mq "db2s/topic-mq"
	"errors"
	"fmt"
	"reflect"
	"strings"
	"sync/atomic"
	"time"
)

func podRowsInit() outPut.Sync {
	var selectRows, insertRows, syncSum int64 = 0, 0, 0
	return outPut.Sync{
		SelectRows: &selectRows, InsertRows: &insertRows, SyncSum: &syncSum,
	}
}

// table sum pod Init
func (t CheckSchemaMetaData) podInit(seq int, v arg.InputTableAttributes) (result *outPut.TPod, err error) {
	var status int64
	result = &outPut.TPod{Seq: seq, Schema: v.Schema, Table: v.Table, Options: v.TableAttributes.Role, CheckMode: GlobalPConfigs.rules.CheckMode, IndexName: "missIndex", IndexType: "none", IndexCol: "none", Status: &status, Result: "ok"}
	switch GlobalPConfigs.rules.CheckMode {
	case "sync":
		result.Sync = podRowsInit()
	}
	return
}

func truncate(partitionGather []string, plan2 *SchedulePlan2) (err error) {
	var event = "[truncate]"
	defer func() {
		if r := ref.RecoverPanic(event, recover()); err == nil && r != nil {
			err = r
		}
	}()
	fmt.Println(fmt.Sprintf("-- [%v] %v begin truncate table %s.%s --", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName, plan2.Object.Schema, plan2.Object.Table))
	c := Meta.NewMeta([]string{GlobalPConfigs.dSns.DestDBName}, "right", "truncate")
	plan2.DstParameter.Object.TruncatePartition = partitionGather
	c.Parameter1 = append(c.Parameter1, plan2.DstParameter)
	if err = c.DDLObjectDrop(); err != nil {
		err = ref.ErrAddPrintf(event, err)
		return
	}
	return
}
func rename(plan *SchedulePlan2) (err error) {
	var (
		currentTime  = time.Now()
		event        = "[rename]"
		recoverTable = fmt.Sprintf("%v.%v", global.RecoverActive, fmt.Sprintf("%v@%v_%v", plan.Object.Schema, plan.Object.MTable, currentTime.Format("20060102150401")))
	)
	defer func() {
		if r := recover(); r != nil {
			err = recoverCost(event, r)
		}
	}()
	fmt.Println(fmt.Sprintf("-- [%v] %v begin rename table %s.%s to %v.%v --", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName, plan.Object.Schema, plan.Object.Table, global.RecoverActive, recoverTable))
	c := Meta.NewMeta([]string{GlobalPConfigs.dSns.DestDBName}, "right", "recover")
	c.Parameter1 = append(c.Parameter1, plan.DstParameter)
	if err = c.DDLObjectDrop(); err != nil {
		err = ref.ErrAddPrintf(event, err)
		return
	}
	return
}

// writeOptimizer 目标端写入的相关优化
func writeOptimizer(plan *SchedulePlan2) {
	var (
		thread int64 = 1
		event        = "[writeOptimizer]"
		err    error
		result any
	)
	plan.writeOptimizer.Thread = &thread
	s := full.NewFull([]string{GlobalPConfigs.dSns.SrcDBName}, "left", plan.Object.Role)
	s.Parameter1 = append(s.Parameter1, plan.SrcParameter)
	if result, err = s.GetFullAvgRowsSizeST(); err != nil || result == nil {
		log.MainLog().Warn(fmt.Sprintf("(%v) %v dest write optimizer fail. error is {%v}. it will use default value.", plan.Object.ObjectLogSeq, event, err))
		plan.writeOptimizer.FixSum = 10
		plan.ReadOptimizer.ChunkSum = 10
		return
	}
	var b int64
	if result.(int64) > 0 {
		b = 16 * 1024 / result.(int64)
	}
	plan.writeOptimizer.AvgRowsSize = result.(int64)
	if b > 0 {
		plan.writeOptimizer.FixSum = b
		plan.ReadOptimizer.ChunkSum = b
		log.MainLog().Info(fmt.Sprintf("(%v) %v dest write optimizer chunkSum is {%v},avg length result is {%v}", plan.Object.ObjectLogSeq, event, b, result))
	} else {
		plan.writeOptimizer.FixSum = 10
		plan.ReadOptimizer.ChunkSum = 10
		log.MainLog().Warn(fmt.Sprintf("(%v) %v dest chunk sum result is 0.avg length result is {%v}, it will use default value.", plan.Object.ObjectLogSeq, event, result))
	}
}

// readOptimizer 目标端写入的相关优化
func readOptimizer(plan2 *SchedulePlan2) {
	var thread int64 = 1
	plan2.ReadOptimizer.ChunkStartSource = "src"
	plan2.ReadOptimizer.Thread = &thread
}


func getTableCol(re any) (srcCol, dstCol []parDef.ColMetaMapS) {
	switch re.(type) {
	case []any:
		if re.([]any)[0] != nil {
			for _, v := range re.([]any)[0].([]parDef.ColMetaMapS) {
				srcCol = append(srcCol, v)
			}
		}
		if re.([]any)[1] != nil {
			for _, v := range re.([]any)[1].([]parDef.ColMetaMapS) {
				dstCol = append(dstCol, v)
			}
		}
	}
	return
}
func getTableColName(qq []parDef.ColMetaMapS) (result []string) {
	for _, v := range qq {
		result = append(result, v.ColumnName)
	}
	return
}
func tableMetaArr1(plan *SchedulePlan2) bool {
	sNew, dNew := getTableCol(plan.baseMe.Col)
	add, eq, del := ea.CheckSumTypeStruct{}.LowerArrCmp(getTableColName(sNew), getTableColName(dNew))
	if !GlobalPConfigs.rules.Sync.IgnoreColumnLength && (len(add) > 0 || len(del) > 0) {
		return false
	}
	if len(eq) == 0 {
		return false
	}
	if len(add) == 0 && len(del) == 0 {
		return true
	}
	return true
}
func tableStatusInit(plan2 *SchedulePlan2) {
	var (
		errEnd int64 = 0
		logSeq int64
	)
	plan2.Status.LogSeq = logSeq
	plan2.Status.ErrEnd = &errEnd
}
func initXlsResult(s *SchemaInitContext) *outPut.WorkSheetResult {
	return &outPut.WorkSheetResult{
		Data: outPut.SheetResultData{
			Seq:    s.Seq,
			Schema: s.Schema,
			Table:  s.Table,
			Role:   s.Role,
			Struct: outPut.StructResultSheet{},
		},
	}
}
func initXlsMiss(s *SchemaInitContext) *outPut.WorkSheetMiss {
	return &outPut.WorkSheetMiss{
		Data: outPut.SheetMiss{
			Schema: s.Schema,
			Table:  s.Table,
			Role:   s.Role,
		},
	}
}
func initSendMsg(plan *SchedulePlan2) {
	plan.MQ.SendMsg = mq.ChanSendMsg{
		Table:             plan.Object.Table,
		MsgData:           plan.MQ.SendMsg.BaseMsg,
		MsgSuccessfulFunc: SendMsgExecSumFunc([]any{plan.MQ.FirstMsg.ProductMsg, int64(1)}),
		MsgFinishFunc:     SendMsgExecSumFunc([]any{[]any{plan.MQ.FirstMsg.CustomerObject}, int64(1)}),
		MsgProducer:       producer,
	}
}
func notSupportColumnType(plan *SchedulePlan2) bool {
	sNew, _ := getTableCol(plan.baseMe.Col)
	for _, v := range sNew {
		if strings.EqualFold(v.TypeBelong, "gis") { //oracle
			return false
		}
	}
	return false
}
func initSyncSchedulePlanObject(s *SchemaInitContext) Object {
	return Object{TaskSeq: s.Seq, ObjectLogSeq: s.ObjectLogSeq, Schema: s.Schema, Table: s.Table, MSchema: s.MSchema, MTable: s.MTable,
		Role: s.Role,
	}
}
func getInsertPre(plan *SchedulePlan2) (res string, err error) {
	var (
		event  = "[getInsertPre]"
		result any
	)
	defer func() {
		if r := ref.RecoverPanic(event, recover()); err == nil && r != nil {
			err = r
		}
	}()
	var parameter = plan.DstParameter
	switch plan.baseMe.Col.(type) {
	case []any:
		parameter.Meta = plan.baseMe.Col.([]any)[1]
	default:
		err = ref.ErrAddPrintf(event, errors.New(fmt.Sprintf("type not match. curry type is %v", reflect.TypeOf(plan.baseMe.Col))))
		return
	}
	c := full.NewFull([]string{GlobalPConfigs.dSns.DestDBName}, "right", plan.Object.Role)
	c.Parameter1 = append(c.Parameter1, parameter)
	if result, err = c.GetFullPrefixST(); err != nil {
		err = ref.ErrAddPrintf(event, err)
		return
	}
	res = fmt.Sprintf("%v", result)
	log.MainLog().Info(fmt.Sprintf("(%v) %v The insert into prefix statement generated in Table %v is {%v}", plan.Object.ObjectLogSeq, event, getTableName(plan.Object.Schema, plan.Object.Table), res))
	return
}
func optionsInit(s *SchemaInitContext) parDef.Options {
	return parDef.Options{
		Scn: GlobalPConfigs.rules.Sync.Scn,
		//StopTime: time.Duration(GlobalPConfigs.rules.Sync.),
		WhereSql:     s.TableAttributes.WhereSql,
		WhereAdd:     Fp.ParseWhereClause(s.TableAttributes.WhereAdd),
		SqlMode:      GlobalPConfigs.rules.Sync.SqlMode,
		TablePx:      s.TableAttributes.PX,
		SqlLogBin:    GlobalPConfigs.rules.Sync.SqlLogBin,
		RowsLimit:    GlobalPConfigs.rules.Sync.RowsLimit,
		RepairMethod: GlobalPConfigs.repair.DataFix,
	}
	//return s.Options
}

func colLobTypeIf(col any) bool {
	switch col.(type) {
	case []parDef.ColMetaMapS:
		for _, v := range col.([]parDef.ColMetaMapS) {
			switch {
			case strings.EqualFold(v.TypeBelong, "lob"):
				return true
			case strings.EqualFold(v.TypeBelong, "longtext"):
				return true
			}
		}
	}
	return false
}
func srcParameterInit(s *SchemaInitContext, recoverTable string) parDef.Parameter {
	var sourceColMeta any
	switch s.BaseMe.Col.(type) {
	case []any:
		sourceColMeta = s.BaseMe.Col.([]any)[0]
	}
	return parDef.Parameter{
		Con: GlobalPConfigs.db.Source.GetDB("single"),
		Object: parDef.Object{Schema: s.Schema, Table: s.Table, RecoverSchema: global.RecoverActive,
			RecoverTable: recoverTable, LobLogo: colLobTypeIf(sourceColMeta),
			Index: func() (res parDef.IndexColumnMe) {
				if s.BaseMe.Index == nil {
					return
				}
				return s.BaseMe.Index.(parDef.IndexColumnMe)
			}(),
			IndexCol: func() (result []string) {
				if s.BaseMe.Index == nil {
					return
				}
				for _, v := range s.BaseMe.Index.(parDef.IndexColumnMe).ColumnMate {
					result = append(result, v.ColumnName)
				}
				return
			}(),
			LockCol: parDef.Encryption{
				Schema:     s.Schema,
				LockFunc:   s.TableAttributes.SecColumn.LockFuncName,
				UnlockFunc: s.TableAttributes.SecColumn.UnLockFuncName,
				Column:     s.TableAttributes.SecColumn.ColumnSlice,
			},
		},
		Options: optionsInit(s),
	}
}
func dstParameterInit(s *SchemaInitContext, recoverTable string) parDef.Parameter {
	var destColMeta any
	switch s.BaseMe.Col.(type) {
	case []any:
		destColMeta = s.BaseMe.Col.([]any)[1]
	}
	return parDef.Parameter{
		Con: GlobalPConfigs.db.Target.GetDB("single"),
		Object: parDef.Object{Schema: s.MSchema, Table: s.MTable,
			RecoverSchema: global.RecoverActive, RecoverTable: recoverTable,
			LobLogo: colLobTypeIf(destColMeta),
			Column: func() (res []parDef.ColMetaMapS) {
				if destColMeta == nil {
					return
				}
				return destColMeta.([]parDef.ColMetaMapS)
			}(),
			LockCol: parDef.Encryption{
				LockFunc:   s.TableAttributes.SecColumn.LockFuncName,
				UnlockFunc: s.TableAttributes.SecColumn.UnLockFuncName,
				Column:     s.TableAttributes.SecColumn.ColumnSlice,
			},
		},
		Options:    optionsInit(s),
		ExecInsert: parDef.InsertSql{},
	}
}
func getColNameSpecifiedFilter(old []parDef.ColMetaMapS, specified []string) (new []parDef.ColMetaMapS) {
	var ll = make(map[string]int)
	for _, v := range specified {
		ll[strings.ToLower(v)]++
	}
	for _, v := range old {
		if _, ok := ll[strings.ToLower(v.ColumnName)]; ok {
			new = append(new, v)
		}
	}
	return
}
func newSyncSchedulePlan(s *SchemaInitContext) (plan2 *SchedulePlan2) {
	var jhSum int64 = 0
	var recoverTable = fmt.Sprintf("%v@%v_%v", s.Schema, s.MTable, time.Now().Format("20060102150401"))
	plan2 = &SchedulePlan2{
		Object:          initSyncSchedulePlanObject(s),
		SrcParameter:    srcParameterInit(s, recoverTable),
		DstParameter:    dstParameterInit(s, recoverTable),
		XlsResult:       initXlsResult(s),
		XlsMiss:         initXlsMiss(s),
		baseMe:          s.BaseMe,
		XlsTerminal:     &outPut.WorkSheetTerminal{},
		TPod:            s.TPod,
		sync:            global.SyncPlan{CheckMod: "sync", CheckRows: global.SyncCheckCount{SrcCheckCount: &jhSum}},
		TableAttributes: s.TableAttributes,
	}
	tableStatusInit(plan2)
	sNew, dNew := getTableCol(s.BaseMe.Col)
	add, eq, del := ea.CheckSumTypeStruct{}.LowerArrCmp(getTableColName(sNew), getTableColName(dNew))
	if !GlobalPConfigs.rules.Sync.IgnoreColumnLength && (len(add) > 0 || len(del) > 0) {
		plan2.baseMe.Col = []any{sNew, dNew}
		plan2.SrcParameter.Object.Column = sNew
		return
	} else if GlobalPConfigs.rules.Sync.IgnoreColumnLength && len(eq) == 0 {
		plan2.baseMe.Col = []any{sNew, dNew}
		plan2.DstParameter.Object.Column = dNew
		return
	}
	newSNew := getColNameSpecifiedFilter(sNew, eq)
	newDNew := getColNameSpecifiedFilter(dNew, eq)
	var virtualCol []string
	for _, v := range newDNew {
		if v.Virtual.Exist {
			virtualCol = append(virtualCol, v.ColumnName)
		}
	}
	newEq, _, _ := ea.CheckSumTypeStruct{}.LowerArrCmp(getTableColName(newSNew), virtualCol)
	plan2.SrcParameter.Object.Column = getColNameSpecifiedFilter(newSNew, newEq)
	plan2.DstParameter.Object.Column = getColNameSpecifiedFilter(newDNew, newEq)
	plan2.DstParameter.ExecInsert.QueryCol = getColNameSpecifiedFilter(newSNew, newEq)
	plan2.DstParameter.ExecInsert.Columns = getColNameSpecifiedFilter(newDNew, newEq)
	plan2.baseMe.Col = []any{plan2.SrcParameter.Object.Column, plan2.DstParameter.Object.Column}
	return
}
func syncSchedulePlanTableMetaInit(plan2 *SchedulePlan2) {
	if !tableMetaArr1(plan2) {
		atomic.SwapInt64(plan2.Status.ErrEnd, -1)
		plan2.Status.ErrInfo = "error(column != length)"
		plan2.TPod.Result = "error(column != length)"
		atomic.SwapInt64(plan2.TPod.Status, -1)
	}
	if notSupportColumnType(plan2) {
		atomic.SwapInt64(plan2.Status.ErrEnd, -1)
		plan2.Status.ErrInfo = "error(notSupportColumnType)"
		plan2.TPod.Result = "error(notSupportColumnType)"
		atomic.SwapInt64(plan2.TPod.Status, -1)
	}
}
func partLikeFilter(forcePartName string, eq []string) (res []string) {
	for _, v := range eq {
		v = strings.ToLower(v)
		forcePartName = strings.ToLower(forcePartName)
		switch {
		case strings.HasPrefix(forcePartName, "%") && !strings.HasSuffix(forcePartName, "%"): //%part
			if strings.HasSuffix(v, strings.ReplaceAll(forcePartName, "%", "")) {
				res = append(res, v)
			}
		case !strings.HasPrefix(forcePartName, "%") && strings.HasSuffix(forcePartName, "%"): //part%
			if strings.HasPrefix(v, strings.ReplaceAll(forcePartName, "%", "")) {
				res = append(res, v)
			}
		case strings.HasPrefix(forcePartName, "%") && strings.HasSuffix(forcePartName, "%"): //%part%
			if strings.Contains(v, strings.ReplaceAll(forcePartName, "%", "")) {
				res = append(res, v)
			}
		case strings.Contains(forcePartName, "%"): //part%01
			var before, after string
			if n := strings.Index(forcePartName, "%"); n != -1 {
				before = forcePartName[:n]
				after = forcePartName[n+1:]
			}
			if strings.Contains(v, before) && strings.Contains(v, after) {
				res = append(res, v)
			}
		default:
			if strings.EqualFold(forcePartName, v) {
				res = append(res, v)
			}
		}
	}
	return
}
func getSrcDstPartMeta(s *SchemaInitContext) (srcPart, dstPart parDef.PartMetaData) {
	if s.BaseMe.Part == nil {
		return
	}
	for k, v := range s.BaseMe.Part.([]any) {
		switch v.(type) {
		case parDef.PartMetaData:
			if k == 0 {
				srcPart = v.(parDef.PartMetaData)
			} else if k == 1 {
				dstPart = v.(parDef.PartMetaData)
			}
		}
	}
	return
}
func ignorePartGarth(partName []string, ignorePartName []string) (qq []string) {
	add, _, del := ea.CheckSum().LowerArrCmp(partName, ignorePartName)
	for _, v := range partName {
		for _, addV := range add {
			if strings.EqualFold(addV, v) {
				qq = append(qq, addV)
			}
		}
		for _, delV := range del {
			if strings.EqualFold(delV, v) {
				qq = append(qq, delV)
			}
		}
	}
	return
}
func partMetaInit(s *SchemaInitContext, plan2 *SchedulePlan2) {
	var forcePartName = strings.TrimSpace(s.TableAttributes.ForcePart)
	//var specifiedPg = s.TableAttributes.PartitionGather
	var SpecifiedPartition, IgnorePartition = s.TableAttributes.SpecifiedPartition, s.TableAttributes.IgnorePartition
	srcPart, dstPart := getSrcDstPartMeta(s)
	_, eq, _ := ea.CheckSum().LowerArrCmp(srcPart.PartName, dstPart.PartName)
	switch {
	case (forcePartName == "normal") || (srcPart.Meth.First != dstPart.Meth.First) || (len(eq) == 0):
		plan2.Object.PartName = []string{"single"}
	case len(forcePartName) > 0:
		_, plan2.Object.PartName, _ = ea.CheckSum().LowerArrCmp(srcPart.PartName, []string{forcePartName})
	case len(SpecifiedPartition) > 0:
		for _, v := range SpecifiedPartition {
			plan2.Object.PartName = append(plan2.Object.PartName, partLikeFilter(v, eq)...)
		}
	default:
		for _, c := range eq {
			plan2.Object.PartName = append(plan2.Object.PartName, partLikeFilter(c, srcPart.PartName)...)
		}
	}
	switch {
	case len(IgnorePartition) > 0:
		plan2.Object.PartName = ignorePartGarth(plan2.Object.PartName, IgnorePartition)
	}
	if len(plan2.Object.PartName) == 0 {
		plan2.Object.PartName = []string{"single"}
	}
}
func initSyncSchedulePlan(s *SchemaInitContext) (plan2 *SchedulePlan2) {
	plan2 = newSyncSchedulePlan(s)
	partMetaInit(s, plan2)
	syncSchedulePlanTableMetaInit(plan2)
	writeOptimizer(plan2)
	readOptimizer(plan2)
	plan2.MQ = mq.NewMonitorMsgMq()
	plan2.MQ.Speed = speedInit(plan2)
	initSendMsg(plan2)
	subTaskInit(plan2)
	if s.TPod.TableMiss {
		plan2.Object.TableMiss = true
	}
	return
}

func InitToSchedulePlan2TableExecFunc(event string, f any) (r any) {
	_, chanParameter, _ := schemaMetaExecFuncParameterSplit(f.([]any))
	if funcName, ok := ScheduleTaskInitFuncRef[GlobalPConfigs.rules.CheckMode]; ok {
		if funcName.Kind() == reflect.Func {
			arguments := []reflect.Value{
				reflect.ValueOf(chanParameter)}
			m1 := funcName.Call(arguments)
			var m2 *SchedulePlan2
			for i := 0; i < len(m1); i++ {
				m2 = m1[i].Interface().(*SchedulePlan2)
			}
			ChanSchedulePlanTable <- m2
		}
	}
	return
}
func InitToSchedulePlan2TableFinishFunc(event string, _ any) (r any) {
	ref.MyWaitDel("scheduleObject")
	close(ChanSchedulePlanTable)
	return
}
func plan2TableScheduleObjectInit(f2 *SchemaInitContext, ChanSchedulePlanTable any) (send *mq.MonitorMsgMq) {
	ref.MyWaitAdd("scheduleObject")
	send = mq.NewMonitorMsgMq()
	send.SendMsg.MsgModifyKey = f2.ChanSchemaContext
	send.NormalQuit = global.EFunc{FuncName: InitToSchedulePlan2TableFinishFunc}
	send.ExecFunc = []global.EFunc{{FuncName: InitToSchedulePlan2TableExecFunc, Params: []any{ChanSchedulePlanTable}}}
	return send
}

// InitToSchedulePlan2Table 初始化 执行计划 以 map 的形式 key 表名 value SchedulePlan2
func InitToSchedulePlan2Table() chan any {
	var f2 = global.GetIoc().GetBean("schemaInitContext").(*SchemaInitContext)
	ChanSchedulePlanTable = make(chan any, f2.TableSum)
	go func (){
		if plan2TableScheduleObjectInit(f2, ChanSchedulePlanTable).NormalChanSendMsg("syR -> func(InitToSchedulePlan2Table)") != nil {
			return
		}
	}()
	return ChanSchedulePlanTable
}
